package com.lzy.canal.service;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.lzy.canal.component.consumer.MysqlConsumer;
import com.lzy.canal.component.result.BaseConsumerResultData;
import com.lzy.canal.component.row.DefaultCanalRow;
import com.lzy.canal.register.consumer.ConsumerRegisterManage;
import lombok.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.ArrayUtils;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author lzy
 */
@Slf4j
public class CanalConsumerService {
    private final ConsumerRegisterManage consumerRegisterManage;

    public CanalConsumerService(ConsumerRegisterManage consumerRegisterManage) {
        this.consumerRegisterManage = consumerRegisterManage;
    }

    public List<? extends BaseConsumerResultData> consumer(CanalEntry.Entry entry) {
        CanalEntry.Header header = entry.getHeader();
        String tableName = header.getTableName();
        String schemaName = header.getSchemaName();
        CanalEntry.RowChange rowChage;
        try {
            rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            log.info("sql-->{}", rowChage.getSql());
        } catch (Exception e) {
            log.error("ERROR ## parser of eromanga-event has an error, data-->{}", entry.getStoreValue(), e);
            throw new RuntimeException(e);
        }
        CanalEntry.EventType eventType = rowChage.getEventType();
        log.info("binlog[{}->{}]", header.getLogfileName(), header.getLogfileOffset());
        log.info("name[{}-->{}]", schemaName, tableName);
        log.info("eventType-->{}", eventType);
        TenantTable tenantTable = convertTableName(tableName);
        MysqlConsumer<? extends BaseConsumerResultData> consumerRegister = consumerRegisterManage.getConsumerRegister(rowChage, schemaName, tenantTable.getTable());
        return consumerRegister.consumer(new DefaultCanalRow(rowChage, tableName));
    }


    private TenantTable convertTableName(String tableName) {
        String[] s = tableName.split("_");
        if (s.length <= 1) {
            return TenantTable.builder().table(tableName).build();
        }
        if (tableName.startsWith("t_")) {
            return getTenantTable(tableName, s, 1);
        }
        return getTenantTable(tableName, s, 0);
    }

    private TenantTable getTenantTable(String tableName, String[] s, int tenantIndex) {
        Long tenantId = getTenantId(s[tenantIndex]);
        if (tenantId != null) {
            Object[] remove = ArrayUtils.remove(s, tenantIndex);
            String table = Stream.of(remove).map(Object::toString).collect(Collectors.joining("_"));
            return TenantTable.builder().table(table).build();
        }
        return TenantTable.builder().table(tableName).build();
    }

    private Long getTenantId(String tenantStr) {
        try {
            return Long.valueOf(tenantStr);
        } catch (Exception e) {
            return null;
        }
    }


    @Data
    @Builder
    @EqualsAndHashCode
    @ToString
    @AllArgsConstructor
    @NoArgsConstructor
    private static class TenantTable {
        /**
         * 表名
         */
        private String table;

    }

}



